package com.seaboxdata.threadHandler;

import com.alibaba.fastjson.JSONObject;
import com.seaboxdata.entity.Warning;
import com.seaboxdata.kafka.BaseStringKafkaConsumer;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import com.seaboxdata.service.WarningService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author
 * @create 2020-11-09 9:20
 **/
@Component
@Slf4j
public class WhpWarningHandler implements DisposableBean, Runnable {

    private Thread thread;

    @Autowired
    private WarningService warningService;

    @Autowired
    private BaseStringKafkaConsumer baseStringKafkaConsumer;

    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    @Value("${whp.warningTopic}")
    private String topic;

    @Value("${interface.isTest}")
    private String isTest;

    private boolean flag;

    @Autowired
    public void initThread() {
        if("true".equals(isTest)){
            return;
        }
        this.flag = true;
        this.thread = new Thread(this);
        thread.setName("whpWarningHandlerThread");
        this.thread.start();
        log.info("whpWarningHandlerThread start");
    }

    @Override
    public void run() {
        boolean success = true;
        KafkaConsumer<String, String> consumer = baseStringKafkaConsumer.getConsumer(topic);
        while (flag) {
            ConsumerRecords<String, String> messages = consumer.poll(Duration.ofMillis(100L));
            List<Warning> results = new ArrayList<>();
            for (ConsumerRecord<String, String> message : messages) {
                results.add(JSONObject.parseObject(message.value(), Warning.class));
            }
            //写入数据库,防止数据在kafka中有重复的,
            if (!CollectionUtils.isEmpty(results)) {
                List<Warning> singleResult = new ArrayList<>();
                for (Warning result : results) {
                    if (!singleResult.contains(result)) {
                        singleResult.add(result);
                    }
                }

                try {
                    List<String> ids = singleResult.stream().map(Warning::getId).collect(Collectors.toList());
                    List<String> existedIds = warningService.selectExistedIds(ids);
                    List<Warning> insertResults = new ArrayList<>();
                    for (Warning warning : singleResult) {
                        if (!existedIds.contains(warning.getId())) {
                            insertResults.add(warning);
                        }
                    }
                    warningService.saveBatch(insertResults);
                    consumer.commitSync();
                } catch (Exception e) {
                    log.error("whp warning commit offset failed");
                   // success = false;
                }

               /* List<String> ids = singleResult.stream().map(Warning::getId).collect(Collectors.toList());
                List<String> existedIds = warningService.selectExistedIds(ids);
                List<Warning> insertResults = new ArrayList<>();
                for (Warning warning : singleResult) {
                    if (!existedIds.contains(warning.getId())) {
                        insertResults.add(warning);
                    }
                }*/

                /*if (success) {
                    try {
                        warningService.saveBatch(insertResults);
                    } catch (Exception e) {
                        log.error("batch insert warning failed");
                        e.printStackTrace();
                        for (Warning warning : insertResults) {
                            try {
                                baseStringKafkaProducer.sendMessageToKafka(JSONObject.toJSONString(warning), topic);
                            } catch (Exception e1) {
                                log.error("write database failed and write kafka warning error");
                                e1.printStackTrace();
                            }
                        }
                    }
                }*/
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        this.flag = false;
    }
}
